package com.cqwshzj.backgroundjob.job;/**
 * @Description:
 * @Auther: Huang Zhijin
 * @Version: 1.0.0
 * @Date: 2021/4/8 15:16
 */

import com.cqwshzj.common.JobConstant;
import com.cqwshzj.framework.dao.model.JobStrategy;
import com.cqwshzj.framework.dao.vo.JobInfoVo;
import com.cqwshzj.framework.job.AbstractSimpleJob;
import com.cqwshzj.util.CrontabUtil;
import com.cqwshzj.util.SysUtils;
import com.dangdang.ddframe.job.api.ShardingContext;
import com.dangdang.ddframe.job.api.simple.SimpleJob;
import org.apache.commons.lang3.RandomUtils;
import org.apache.commons.lang3.StringUtils;
import org.apache.commons.lang3.time.DateFormatUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.util.Assert;
import org.springframework.util.CollectionUtils;

import java.time.LocalDate;
import java.time.LocalDateTime;
import java.time.LocalTime;
import java.time.format.DateTimeFormatter;
import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.function.Consumer;

/**
 *
 * 扫描 jobInfo 并添加到 zookeeper（elastic-job）的后台任务
 * @author Huang Zhijin
 * @create 2021/4/8
 * @since 1.0.0
 */
public class ScanJobInfoAndAddToZkDaemonJob extends AbstractSimpleJob implements SimpleJob {

	private static final Logger log = LoggerFactory.getLogger( ScanJobInfoAndAddToZkDaemonJob.class );

	//-------------------------  elastic-job-lite  SimpleJob execute method  ---------------------------

	private Map<Integer, Consumer<JobInfoVo>>  jobInfoVoHandlerMap = new HashMap<>();

	private Consumer<JobInfoVo> jobInfoVoExceptionHandlerConsumer = (obj)->{
		try {
			Integer strtType = obj.getJobStrategy() == null || obj.getJobStrategy().getStrtType() == null
					? null : obj.getJobStrategy().getStrtType();
			jobInfoVoHandlerMap.get(strtType).accept(obj);
		}catch (Exception e){
			log.error(" scan jobInfo and reExe job error. jobInfo:[ "+ SysUtils.obj2JsonStr(obj) +" ] . exception: "+e.getMessage(),e);
		}
	};


	private int jobAddedCount = 0;

	//-------------------------  elastic-job-lite  SimpleJob execute method  ---------------------------


	@Override
	public void execute(ShardingContext shardingContext) {
		super.executeJob(shardingContext);
	}

	@Override
	public void initInstanceProperties(ShardingContext shardingContext) {
		super.initInstanceProperties(shardingContext);
		//---------------
		jobInfoVoHandlerMap.put(JobConstant.JOB_STRATEGY_TYPE_ONLY_ONCE , this::reExeOnlyOnce );
		jobInfoVoHandlerMap.put(JobConstant.JOB_STRATEGY_TYPE_ONCE_A_DAY , this::reExeOnlyOnce );
		jobInfoVoHandlerMap.put(JobConstant.JOB_STRATEGY_TYPE_INTERVAL , this::reExeIntervalJob );
		jobInfoVoHandlerMap.put( null , this::noOperation );
	}

	@Override
	public void doExecute(ShardingContext shardingContext) {
		jobAddedCount = 0;
		long start = System.currentTimeMillis();
		log.info(" scan jobInfo start.  startTime=[{}] ",start );


		//--1.扫描待执行且未放入 elastic-job 的任务并动态添加至 elastic-job
		jobAddedCount += jobInfoService.scanJobInfoAndAdd();

		//--2. 扫描曾经  执行失败 或者 执行超时的任务  并重新执行
		final List<JobInfoVo> reExeList = jobInfoDao.listWithStrategy(JobConstant.JOB_STATE_FAILED, JobConstant.JOB_STATE_TIMEOUT);

		if ( !CollectionUtils.isEmpty( reExeList )) {
			reExeList.forEach( jobInfoVoExceptionHandlerConsumer );
		}

		if (log.isInfoEnabled()) {
			log.info(" scan jobInfo successfully. added {} jobs. total cost [{}]ms . ", jobAddedCount, (System.currentTimeMillis()-start) );
		}
	}




	//----------------------------------

	/**
	 * 空实现
	 * @param vo
	 */
	public void noOperation(JobInfoVo vo){
	}



	/**
	 * 近重新执行任务一次
	 * @param vo
	 */
	public void reExeOnlyOnce(JobInfoVo vo){
		Assert.notNull(vo , "缺少参数 任务信息 vo");
		Assert.notNull(vo.getId() , "缺少参数 任务信息 id");

		if (checkJobLockAndJobTypeConcurrencyControl( vo )) {
			log.info("jobInfo( id = {} , jobType = {} ) re-execute error ! reach max concurrency limit or add lock failed. skip re-execute current job. ",vo.getId(),vo.getJobType());
			return;
		}

		JobStrategy jobStrategy = vo.getJobStrategy();
		Integer strtType = jobStrategy == null ? null : jobStrategy.getStrtType();
		if (strtType == null) {
			LocalDateTime jobExeDateTime = LocalDateTime.now().plusSeconds(RandomUtils.nextInt(2, JobConstant.DELAY_SECONDS_BEGIN));
			String cron = CrontabUtil.getCrontabByLocalDateTime( jobExeDateTime );
			jobInfoService.reExeJobByIdAndCron( vo.getId(), cron );
			return;
		}

		Assert.notNull( jobStrategy , "缺少参数 任务策略信息 jobStrategy");
		Assert.notNull( jobStrategy.getStartTime(), "缺少参数 任务策略的开始时间 startTime ");

		DateTimeFormatter timeFormatter = DateTimeFormatter.ofPattern( JobConstant.TIME_4 );
		DateTimeFormatter dateTimeFormatter = DateTimeFormatter.ofPattern( JobConstant.DATETIME_3 );

		LocalDate today = LocalDate.now();
		LocalTime jobStartTime = LocalTime.parse( jobStrategy.getStartTime() , timeFormatter );
		LocalTime jobEndTime = jobStartTime.plusMinutes( jobStrategy.getTimeWindow() != null ? 60 : jobStrategy.getTimeWindow() );

		LocalDateTime jobStartDateTime = LocalDateTime.of( today, jobStartTime );
		LocalDateTime jobEndDateTime = LocalDateTime.of( today, jobEndTime );

		LocalDateTime now = LocalDateTime.now();
		LocalDateTime jobExeDateTime = null;
		// now < jobStartDateTime
		if ( now.isBefore( jobStartDateTime ) ){
			jobExeDateTime = jobStartDateTime;
		}
		// jobStartDateTime <= now < jobEndDateTime
		else if( !now.isBefore(jobStartDateTime) && now.isBefore( jobEndDateTime) ){
			jobExeDateTime = now.plusSeconds( RandomUtils.nextInt(2, JobConstant.DELAY_SECONDS_BEGIN) );
		}
		else if( now.isAfter( jobEndDateTime) || now.isEqual( jobEndDateTime) ){
			log.debug(" jobInfo( jobId="+vo.getId()+") of currentTime( "+ DateFormatUtils.format(new Date(),JobConstant.DATETIME_3) +") is not in time windows( "+jobStartDateTime.format(dateTimeFormatter)+" , "+jobEndDateTime.format(dateTimeFormatter)+" ) . skip re-execute current job.");
		}
		if (jobEndDateTime != null) {
			String cron = CrontabUtil.getCrontabByLocalDateTime( jobExeDateTime );
			jobInfoService.reExeJobByIdAndCron( vo.getId(), cron );
			jobAddedCount++;
		}
	}


	/**
	 *
	 * @param vo
	 */
	public void reExeIntervalJob(JobInfoVo vo){
		Assert.notNull( vo , "缺少参数 任务信息 vo");
		Assert.notNull( vo.getId() , "缺少参数 任务信息 id");
		Assert.notNull( vo.getJobStrategy() , "缺少参数 任务信息 id");

		if (checkJobLockAndJobTypeConcurrencyControl(vo)) {
			return;
		}

		JobStrategy jobStrategy = vo.getJobStrategy();
		Assert.isTrue( jobStrategy.getStrtType() != null && jobStrategy.getStrtType().intValue() == JobConstant.JOB_STRATEGY_TYPE_INTERVAL, "缺少 strtType 或 strtType 不等于 3");
		Assert.hasText( jobStrategy.getIntvMin(), "缺少 任务策略的执行周期分钟数 intvMin ");
		Assert.isTrue( jobStrategy.getTimeWindow() != null && jobStrategy.getTimeWindow().intValue() >0 , "缺少任务策略的任务执行窗口其 ");

		DateTimeFormatter dateTimeFormatter = DateTimeFormatter.ofPattern( JobConstant.JOB_dATE );
		DateTimeFormatter timeFormatter = DateTimeFormatter.ofPattern( JobConstant.TIME_4 );

		LocalDate today = LocalDate.now();
		LocalTime jobStartTime = LocalTime.parse( jobStrategy.getStartTime(), timeFormatter );
		LocalTime jobEndTime = jobStartTime.plusMinutes( jobStrategy.getTimeWindow() == null ? 60 : jobStrategy.getTimeWindow() );

		LocalDateTime jobStartDateTime = LocalDateTime.of(today, jobStartTime );
		LocalDateTime jobEndDateTime = LocalDateTime.of(today, jobEndTime );

		LocalDateTime now = LocalDateTime.now();
		LocalDateTime jobLastUpdTime = StringUtils.isEmpty( vo.getJobUpdTime() ) ? now : LocalDateTime.parse( vo.getJobUpdTime() , dateTimeFormatter );

		if (now.isAfter( jobStartDateTime ) && now.isBefore( jobEndDateTime )) {
			log.info(" current interval job is still in today's time-window . framework can not re-execute current job.");
			return;
		}
		// 日期加 1 天，并重新执行
		LocalDateTime jobExeStartTime = jobStartDateTime.plusDays(1L);
		LocalDateTime jobExeEndTime = jobEndDateTime.plusDays(1L);
		int intervalMin = Integer.parseInt( jobStrategy.getIntvMin().trim() );
		int hourBegin = jobStartDateTime.getHour();
		int hourEnd = jobStartDateTime.getHour();

		LocalDate tomorrow = today.plusDays(1L);
		String intervalCron = CrontabUtil.getIntervalMinuteCrontabOfTargetDate(tomorrow, intervalMin, hourBegin, hourEnd );

		jobInfoService.reExeJobByIdAndCron(vo.getId(), intervalCron);
		jobAddedCount++;
	}



	private boolean checkJobLockAndJobTypeConcurrencyControl(JobInfoVo vo){
		try {
			jobInfoService.lockAndUpdateUpdTimeById(vo.getId(), vo.getJobUpdTime() );
		}catch (Exception e){
			log.info("jobInfo( id = "+vo.getId()+" ), add lock failed. skip re-execute current job . exception message: "+e.getMessage());
		}

		boolean control = jobInfoService.isConcurrencyControlByJobType(vo.getJobType() == null ? 0 : vo.getJobType().intValue() );
		if (control) {
			log.info("jobInfo( id = {} , jobType = {} ) reach max concurrency limit . skip re-execute current job. ",vo.getId(),vo.getJobType());
			return true;
		}
		return false;
	}

}
